package com.example.demo.config;

import com.example.demo.constant.RqConstant;
import com.example.demo.hello.UserReceiver;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: anlu
 * @Date: 2021/06/23
 * @Description:
 */
@Configuration
public class RabbitConfig {
    @Value("${spring.rabbitmq.host}")
    private String addresses;

    @Value("${spring.rabbitmq.port}")
    private String port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

//    @Value("${spring.rabbitmq.publisher-confirms}")
//    private boolean publisherConfirms;

    @Autowired
    private UserReceiver userReceiver;

    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(addresses+":"+port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);

//        connectionFactory.setPublisherConfirms(true);
        /** 如果要进行消息回调，则这里必须要设置为true */
//        connectionFactory.setPublisherConfirms(publisherConfirms);
        // 开启生产者确认模式 Correlated 值是发布消息成功到交换器后会触发回调方法
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory){
        return new RabbitAdmin(connectionFactory);
    }
    @Bean
    public RabbitTemplate newRabbitTemplate(){
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMandatory(true);
        // 设置生产者确认模式的回调方法
        template.setConfirmCallback(confirmCallback());
        //设置Mandatory的失败通知，需要和ConfirmType.CORRELATED配合使用
        template.setReturnCallback(returnCallback());
        return template;

    }

    //===============使用了RabbitMQ系统缺省的交换器==========
    //绑定键即为队列名称
    @Bean
    public Queue helloQueue() {
        return new Queue(RqConstant.QUEUE_HELLO);
    }
    @Bean
    public Queue userQueue() {
        return new Queue(RqConstant.QUEUE_USER);
    }

    //===============以下是验证topic Exchange==========
    @Bean
    public Queue queueEmailMessage() {
        return new Queue(RqConstant.QUEUE_TOPIC_EMAIL);
    }
    @Bean
    public Queue queueUserMessages() {
        return new Queue(RqConstant.QUEUE_TOPIC_USER);
    }

    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(RqConstant.EXCHANGE_TOPIC);
    }

    @Bean
    public Binding bindingEmailExchangeMessage() {
        return BindingBuilder
                .bind(queueEmailMessage())
                .to(exchange())
                .with("an.*.email");

    }
    @Bean
    public Binding bindingUserExchangeMessages() {
        return BindingBuilder
                .bind(queueUserMessages())
                .to(exchange())
                .with("an.*.user");
    }

    //===============以下是验证Fanout Exchange==========
    @Bean
    public Queue AMessage() {
        return new Queue("an.fanout.A");
    }

    @Bean
    public Queue BMessage() {
        return new Queue("an.fanout.B");
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(RqConstant.EXCHANGE_FANOUT);
    }

    @Bean
    Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(Queue BMessage,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }

    @Bean
    public SimpleMessageListenerContainer messageListenerContainer(){
        SimpleMessageListenerContainer container
                = new SimpleMessageListenerContainer(connectionFactory());
        container.setQueues(userQueue());
        // 设置手动确认模式
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(userReceiver);
        return container;
    }


    //============生产者发送确认============
    /**
     * 这个是开启发送者确认模式的回调方法
     * @param
     * @author: anlu
     * @time: 2021/7/8
     */
    @Bean
    public RabbitTemplate.ConfirmCallback confirmCallback(){
        return new RabbitTemplate.ConfirmCallback(){

            @Override
            public void confirm(CorrelationData correlationData,
                                boolean ack, String cause) {
                if (ack) {
                    System.out.println("==confirmCallback：发送者确认发送给mq成功");
                } else {
                    //处理失败的消息
                    System.out.println("==发送者发送给mq失败,考虑重发:"+cause);
                }
            }
        };
    }

    /**
     * 这个是Mandatory回调的方法
     * @param
     * @author: anlu
     * @time: 2021/7/8
     */
    @Bean
    public RabbitTemplate.ReturnCallback returnCallback(){
        return new RabbitTemplate.ReturnCallback(){

            @Override
            public void returnedMessage(Message message,
                                        int replyCode,
                                        String replyText,
                                        String exchange,
                                        String routingKey) {
                System.out.println("----------------------");
                System.out.println("无法路由的消息，需要考虑另外处理。");
                System.out.println("Returned replyText："+replyText);
                System.out.println("Returned exchange："+exchange);
                System.out.println("Returned routingKey："+routingKey);
                String msgJson  = new String(message.getBody());
                System.out.println("Returned Message："+msgJson);
                System.out.println("----------------------");
            }
        };
    }


}
